package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.TestExamBean;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import groupOne.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * use edu;
 * drop table if exists dws_test_user_test_exam_window;
 * create table if not exists dws_test_user_test_exam_window(
 *     stt datetime,
 *     edt datetime,
 *     cur_date date,
 *     paper_id varchar(20) comment '考卷id',
 *     paper_title varchar(500) comment '考卷名称',
 *     course_id varchar(20)  comment '课程id',
 *     course_name varchar(500) comment '课程名称',
 *     total_score double replace comment '总分数',
 *     uu_ct bigint replace comment '考试人数',
 *     total_dur_sec bigint replace comment '考试时长'
 * )engine = olap
 * aggregate key(`stt`,`edt`,`cur_date`,`paper_id`,`paper_title`,`course_id`,`course_name`)
 * comment"考试域测试汇总表"
 * partition by range(`cur_date`)()
 * distributed by hash(stt) buckets 10 properties(
 *   "replication_num" = "3",
 *   "dynamic_partition.enable" = "true",
 *   "dynamic_partition.time_unit" = "DAY",
 *   "dynamic_partition.start" = "-1",
 *   "dynamic_partition.end" = "1",
 *   "dynamic_partition.prefix" = "par",
 *   "dynamic_partition.buckets" = "10",
 *   "dynamic_partition.hot_partition_num" = "3"
 * );
 */
public class Dws_01_TextUserTestExamXiang extends BaseAppOneStream {
    public static void main(String[] args) {
        new Dws_01_TextUserTestExamXiang().init(
                6003,
                2,
                "Dws_01_TextUserTestCourseExamXiang",
                Constant.TOPIC_DWD_TEST_TEST_EXAM
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //封装pojo
        SingleOutputStreamOperator<TestExamBean> pojoStream = transFunc(stream);
        //聚合
        SingleOutputStreamOperator<TestExamBean> aggStream = aggFunc(pojoStream);
        //add dim
        SingleOutputStreamOperator<TestExamBean> resultStream = addDim(aggStream);
        //写入doris
//        resultStream.print();
        writeDoris(resultStream);

    }

    private void writeDoris(SingleOutputStreamOperator<TestExamBean> resultStream) {
        resultStream
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                    return JSON.toJSONString(bean, config);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_test_user_test_exam_window"));
    }

    private SingleOutputStreamOperator<TestExamBean> addDim(SingleOutputStreamOperator<TestExamBean> aggStream) {
        SingleOutputStreamOperator<TestExamBean> paperStream = AsyncDataStream.unorderedWait(
                aggStream,
                new DimAsyncFunction<TestExamBean>() {
                    @Override
                    public String getTable() {
                        return "dim_test_paper";
                    }

                    @Override
                    public String getId(TestExamBean input) {
                        return input.getPaperId();
                    }

                    @Override
                    public void addDim(JSONObject dim, TestExamBean bean) {
                        bean.setPaperTitle(dim.getString("PAPER_TITLE"));
                        bean.setCourseId(dim.getString("COURSE_ID"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<TestExamBean> stream = AsyncDataStream.unorderedWait(
                paperStream,
                new DimAsyncFunction<TestExamBean>() {
                    @Override
                    public String getTable() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getId(TestExamBean input) {
                        return input.getCourseId();
                    }

                    @Override
                    public void addDim(JSONObject dim, TestExamBean bean) {
                        bean.setCourseName(dim.getString("COURSE_NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        return stream;
    }

    private SingleOutputStreamOperator<TestExamBean> aggFunc(SingleOutputStreamOperator<TestExamBean> pojoStream) {
        return pojoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TestExamBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((t,l) -> t.getTs())
                )
                .keyBy(TestExamBean::getPaperId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TestExamBean>() {
                            @Override
                            public TestExamBean reduce(TestExamBean bean1, TestExamBean bean2) throws Exception {
                                bean1.setTotalScore(bean1.getTotalScore() + bean2.getTotalScore());
                                bean1.setUuCt(bean1.getUuCt() + bean2.getUuCt());
                                bean1.setTotalDurSec(bean1.getTotalDurSec() + bean2.getTotalDurSec());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TestExamBean, TestExamBean, String, TimeWindow>() {
                            @Override
                            public void process(String paperId,
                                                ProcessWindowFunction<TestExamBean, TestExamBean, String, TimeWindow>.Context ctx,
                                                Iterable<TestExamBean> iterable,
                                                Collector<TestExamBean> collector) throws Exception {
                                TestExamBean bean = iterable.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(ctx.window().getStart()));

                                collector.collect(bean);
                            }
                        });
    }

    private SingleOutputStreamOperator<TestExamBean> transFunc(DataStreamSource<String> stream) {
        return stream
                .map(new MapFunction<String, TestExamBean>() {
                    @Override
                    public TestExamBean map(String json) throws Exception {
                        JSONObject obj = JSON.parseObject(json);
                        String paperId = obj.getString("paper_id");
                        Double score = obj.getDouble("score");
                        Long durationSec = obj.getLong("duration_sec");
                        Long ts = obj.getLong("ts")*1000;
                        return new TestExamBean(
                                "","","",
                                paperId,"","","",
                                score,1l,durationSec,
                                ts
                        );
                    }
                });

    }
}
